package com.mazaiting.rabbitmq.service.impl;

import cn.hutool.core.convert.Convert;
import com.mazaiting.common.core.domain.result.ResultCode;
import com.mazaiting.common.core.ex.exceptions.BaseException;
import com.mazaiting.common.core.ex.exceptions.EmptyObjectException;
import com.mazaiting.rabbitmq.callback.IMessageCallback;
import com.mazaiting.rabbitmq.constant.MqConstant;
import com.mazaiting.rabbitmq.domain.info.RabbitModuleInfo;
import com.mazaiting.rabbitmq.listeners.DynamicMQListener;
import com.mazaiting.rabbitmq.service.IDynamicMQService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
import org.springframework.retry.interceptor.RetryOperationsInterceptor;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 动态 MQ 服务实现类
 *
 * @author mazaiting
 * @create_time 2022/9/28 23:37
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class DynamicMQServiceImpl implements IDynamicMQService {

    /**
     * MQ 模板工具
     */
    private final RabbitTemplate rabbitTemplate;

    /**
     * MQ 管理者
     */
    private final RabbitAdmin rabbitAdmin;

    /**
     * 动态 MQ 监听
     */
    private final DynamicMQListener dynamicMQListener;

    /**
     * 重试操作
     */
    private final RetryOperationsInterceptor retryOperationsInterceptor;

    /**
     * 消息容器工厂
     */
    private final RabbitListenerContainerFactory<DirectMessageListenerContainer> rabbitListenerContainerFactory;

    /**
     * 消息监听容器
     */
    private static final Map<String, DirectMessageListenerContainer> CONTAINER_MAP = new ConcurrentHashMap<>(8);

    @Override
    public boolean add(RabbitModuleInfo rabbitModuleInfo) throws BaseException {
        // 校验参数
        validateParam(rabbitModuleInfo);
        // 生成队列
        Queue queue = convertQueue(rabbitModuleInfo.getQueue());
        // 根据队列名判断队列是否存在, 已存在则返回添加失败
        if (exists(queue.getName())) {
            return false;
        }
        // 交换机
        Exchange exchange = convertExchange(rabbitModuleInfo.getExchange());
        // 路由键
        String routingKey = rabbitModuleInfo.getRoutingKey();
        // 绑定关系
        Binding binding = BindingBuilder.bind(queue).to(exchange).with(routingKey).noargs();
//        Binding binding = new Binding(queueName, Binding.DestinationType.QUEUE, exchangeName, routingKey, null);
        // 创建队列
        String result = rabbitAdmin.declareQueue(queue);
        // 创建失败
        if (!StringUtils.hasLength(result)) {
            return false;
        }
        rabbitAdmin.declareExchange(exchange);
        rabbitAdmin.declareBinding(binding);
        return true;
    }

    @Override
    public boolean deleteQueue(String queueName) {
        // 获取队列信息
        QueueInformation queueInfo = rabbitAdmin.getQueueInfo(queueName);
        if (Objects.isNull(queueInfo)) {
            log.error("删除队列失败. 队列不存在: " + queueName);
            return false;
        }
        // 队列消息数量为空则删除
        int messageCount = queueInfo.getMessageCount();
        if (messageCount == 0) {
            boolean result = rabbitAdmin.deleteQueue(queueName);
            log.info("删除队列{}, 队列名: {}", result ? "成功" : "失败", queueName);
            return result;
        } else {
            log.info("删除队列失败, [{}]里还有[{}]条消息。不做删除操作", queueName, messageCount);
            return false;
        }
    }

    @Override
    public boolean startListener(RabbitModuleInfo rabbitModuleInfo, int consumerNum, IMessageCallback<Object> messageCallback) throws BaseException {
        // 添加队列和交换机
        boolean result = add(rabbitModuleInfo);
        // 获取队列
        RabbitModuleInfo.Queue queue = rabbitModuleInfo.getQueue();
        // 队列名
        String queueName = queue.getName();
        if (!result) {
            log.error("MQ 队列已存在: {}", queueName);
        }
        // 交换机名
        String exchangeName = rabbitModuleInfo.getExchange().getName();
        // 路由 key
        String routingKey = rabbitModuleInfo.getRoutingKey();
        // 判断是否死信是否自动创建容器. 延时队列 = 死信交换机不为空 && 死信路由 key 不为空 && 延时队列
        boolean autoCreate = !(StringUtils.hasLength(queue.getDeadLetterExchange())
                && StringUtils.hasLength(queue.getDeadLetterRoutingKey())
                && queue.isDelay());
        // 非延时队列则创建
        if (Objects.nonNull(messageCallback) && autoCreate) {
            // 线程共享
            DirectMessageListenerContainer container = rabbitListenerContainerFactory.createListenerContainer();
            // 获取消息容器
            DirectMessageListenerContainer directMessageListenerContainer = CONTAINER_MAP.get(queueName);
            // 判断是否已在容器中存在
            if (Objects.nonNull(directMessageListenerContainer)) {
                container = directMessageListenerContainer;
                log.info("动态修改mq监听成功: 交换机-{}, 路由key-{}, 队列-{}, 线程数-{}", exchangeName, routingKey, queueName, consumerNum);
            } else {
                // 设置队列名
                container.setQueueNames(queueName);
                log.info("动态添加mq监听成功: 交换机-{}, 路由key-{}, 队列-{}, 线程数-{}", exchangeName, routingKey, queueName, consumerNum);
            }
            // 消息发送者发送的数量
            container.setPrefetchCount(5);
            // 队列的消费者数量
            container.setConsumersPerQueue(consumerNum);
            // 设置回调
            dynamicMQListener.setMessageCallback(messageCallback);
            // 设置消息监听
            container.setMessageListener(dynamicMQListener);
            // 启动容器监听
            container.start();
            CONTAINER_MAP.put(queueName, container);
        }
        return true;
    }

    @Override
    public boolean stopListener(String queueName) {
        // 监听容器
        DirectMessageListenerContainer container = CONTAINER_MAP.get(queueName);
        if (Objects.nonNull(container)) {
            container.stop();
            container.destroy();
            CONTAINER_MAP.remove(queueName);
        }
        log.info("停止监听队列, 队列名: {}", queueName);
        return deleteQueue(queueName);
    }

    @Override
    public boolean exists(String queueName) {
        return Objects.nonNull(rabbitAdmin.getQueueInfo(queueName));
    }

    /**
     * RabbitMQ动态配置参数校验
     *
     * @param rabbitModuleInfo 模块信息
     */
    public void validateParam(RabbitModuleInfo rabbitModuleInfo) throws EmptyObjectException {
        // 校验模块信息不能为空
        if (Objects.isNull(rabbitModuleInfo)) {
            throw new EmptyObjectException(ResultCode.MQ_MODULE_INFO_EMPTY);
        }

        // 路由键
        String routingKey = rabbitModuleInfo.getRoutingKey();
        // 判断路由 key 是否配置
        if (!StringUtils.hasLength(routingKey)) {
            throw new EmptyObjectException(ResultCode.MQ_ROUTING_KEY_EMPTY);
        }
        // 判断交换机是否为空
        if (Objects.isNull(rabbitModuleInfo.getExchange())) {
            log.error("routingKey: {}未配置exchange", routingKey);
            throw new EmptyObjectException(ResultCode.MQ_EXCHANGE_EMPTY);
        }

        // 交换机的名字是否配置
        if (!StringUtils.hasLength(rabbitModuleInfo.getExchange().getName())) {
            log.error("routingKey: {}未配置exchange的name属性", routingKey);
            throw new EmptyObjectException(ResultCode.MQ_EXCHANGE_NAME_EMPTY);
        }

        // 判断队列是否为空
        if (Objects.isNull(rabbitModuleInfo.getQueue())) {
            log.error("routingKey: {}未配置queue", routingKey);
            throw new EmptyObjectException(ResultCode.MQ_QUEUE_EMPTY);
        }

        // 判断队列名称是否为空
        if (!StringUtils.hasLength(rabbitModuleInfo.getQueue().getName())) {
            log.error("routingKey: {}未配置queue的name属性", routingKey);
            throw new EmptyObjectException(ResultCode.MQ_QUEUE_NAME_EMPTY);
        }
    }


    /**
     * 转换生成RabbitMQ队列
     *
     * @param queue 队列参数
     * @return 消息队列
     */
    public Queue convertQueue(RabbitModuleInfo.Queue queue) {
        // 获取额外参数
        Map<String, Object> arguments = queue.getArguments();

        // 转换ttl的类型为long
        if (Objects.nonNull(arguments) && arguments.containsKey(MqConstant.KEY_MESSAGE_TTL)) {
            arguments.put(MqConstant.KEY_MESSAGE_TTL, Convert.toLong(arguments.get(MqConstant.KEY_MESSAGE_TTL)));
        }

        // 死信交换机
        String deadLetterExchange = queue.getDeadLetterExchange();
        // 死信路由 key
        String deadLetterRoutingKey = queue.getDeadLetterRoutingKey();

        // 路由交换机与路由 key不为空
        if (StringUtils.hasLength(deadLetterExchange) && StringUtils.hasLength(deadLetterRoutingKey)) {
            // 为空则创建
            if (Objects.isNull(arguments)) {
                arguments = new HashMap<>(4);
            }
            // 设置死信参数
            arguments.put(MqConstant.KEY_DEAD_LETTER_EXCHANGE, deadLetterExchange);
            arguments.put(MqConstant.KEY_DEAD_LETTER_ROUTING, deadLetterRoutingKey);
        }
        // 生成队列
        return new Queue(queue.getName(), queue.isDurable(), queue.isExclusive(), queue.isAutoDelete(), arguments);
    }


    /**
     * 转换生成RabbitMQ交换机
     *
     * @param exchangeInfo 交换机信息
     * @return 交换机
     */
    public Exchange convertExchange(RabbitModuleInfo.Exchange exchangeInfo) {
        // 交换机类型
        String type = exchangeInfo.getType().name().toLowerCase();
        // 交换机名
        String exchangeName = exchangeInfo.getName();
        // 额外参数
        Map<String, Object> arguments = exchangeInfo.getArguments();
        // 校验map是否为空
        arguments = Objects.nonNull(arguments) ? arguments : new HashMap<>();
        // 交换机构造器
        ExchangeBuilder builder =
                new ExchangeBuilder(exchangeName, type)
                        // 设置持久化
                        .durable(exchangeInfo.isDurable())
                        // 设置参数
                        .withArguments(arguments);
        // 设置自动删除
        if (exchangeInfo.isAutoDelete()) {
            builder.autoDelete();
        }
        return builder.build();
    }

}
